package com.hrt.iceberg

import org.apache.hadoop.conf.Configuration
import org.apache.iceberg.Table
import org.apache.iceberg.actions.Actions
import org.apache.iceberg.catalog.TableIdentifier
import org.apache.iceberg.hadoop.HadoopCatalog
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * Spark 查询Iceberg，包含：
  *   1.DataFrame API 加载Iceberg 表操作
  *   2.查询表快照
  *   3.查询表历史
  *   4.查询表data files
  *   5.查询Manifests
  *   6.查询指定快照数据
  *   7.根据时间戳查询数据
  *   8.回滚快照
  *   9.合并Iceberg表的数据文件
  *   10.删除历史快照
  *
  */
object SparkIcebergQueries {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local").appName("test")
      //指定hadoop catalog，catalog名称为hadoop_prod
      .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")
      .config("spark.sql.catalog.hadoop_prod.type", "hadoop")
      .config("spark.sql.catalog.hadoop_prod.warehouse", "hdfs://mycluster/sparkoperateiceberg")
      .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
//      .config("spark.sql.autoBroadcastJoinThreshold",-1)
      .getOrCreate()

    //1.创建Iceberg表，并插入数据
//    spark.sql(
//      """
//        |create table hadoop_prod.mydb.mytest (id int,name string,age int) using iceberg
//      """.stripMargin)
//    spark.sql(
//      """
//        |insert into hadoop_prod.mydb.mytest values (1,"zs",18),(2,"ls",19),(3,"ww",20)
//      """.stripMargin)
    //1).SQL 方式读取Iceberg中的数据
//    spark.sql("select * from hadoop_prod.mydb.mytest").show()

    /**
      * 2).使用Spark查询Iceberg中的表除了使用sql 方式之外，还可以使用DataFrame方式,建议使用SQL方式
      */
    //第一种方式使用DataFrame方式查询Iceberg表数据
//    val frame1: DataFrame = spark.table("hadoop_prod.mydb.mytest")
//    frame1.show()
    //第二种方式使用DataFrame加载 Iceberg表数据
//    val frame2: DataFrame = spark.read.format("iceberg").load("hdfs://mycluster/sparkoperateiceberg/mydb/mytest")
//    frame2.show()

    /**
      * 对Iceberg表进行 查询快照、查询表历史、查询Data files、查询Manifests、
      * 查询指定快照数据、根据时间戳查询数据、回滚快照、xxx,xx,xx 操作
     */
    //向表 hadoop_prod.mydb.mytest 中再次插入以下数据
//    spark.sql(
//      """
//        |insert into hadoop_prod.mydb.mytest values (4,"ml",18),(5,"tq",19),(6,"gb",20)
//      """.stripMargin)

    //3.查看Iceberg表快照信息
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.snapshots
//      """.stripMargin).show(false)

    //4.查询表历史,实际上就是表快照的部分内容
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.history
//      """.stripMargin).show(false)

    //5.查看表对应的data files
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.files
//      """.stripMargin).show(false)

    //6.查看表对应的 Manifests
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.manifests
//      """.stripMargin).show(false)

    //7.查询指定快照数据，快照ID可以通过读取json元数据文件获取
//    spark.read
//      .option("snapshot-id",3368002881426159310L)
//      .format("iceberg")
//      .load("hdfs://mycluster/sparkoperateiceberg/mydb/mytest")
//      .show()

    //SQL 方式指定查询快照ID 数据
//    spark.sql(
//      """
//        |call hadoop_prod.system.set_current_snapshot('mydb.mytest',3368002881426159310)
//      """.stripMargin)
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest
//      """.stripMargin).show()

    //8.根据时间戳查询数据,时间戳指定成毫秒，iceberg会根据元数据找出timestamp-ms <= as-of-timestamp 对应的 snapshot-id ，把数据查询出来
//    spark.read.option("as-of-timestamp","1640066148000")
//      .format("iceberg")
//      .load("hdfs://mycluster/sparkoperateiceberg/mydb/mytest")
//      .show()

    //SQL 方式查询指定 时间戳 快照数据
//    spark.sql(
//      """
//        |CALL hadoop_prod.system.rollback_to_timestamp('mydb.mytest', TIMESTAMP '2021-12-23 16:56:40.000')
//      """.stripMargin)
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest
//      """.stripMargin).show()


    //9.回滚到某个快照，rollbackTo(snapshot-id)，指定的是固定的某个快照ID，回滚之后，会生成新的Snapshot-id, 重新查询生效。
//    val conf = new Configuration()
//    val catalog = new HadoopCatalog(conf,"hdfs://mycluster/sparkoperateiceberg")
//    catalog.setConf(conf)
//    val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
//    table.manageSnapshots().rollbackTo(3368002881426159310L).commit()

    //SQL方式回滚快照ID，操作如下：
//    spark.sql(
//      """
//        |Call hadoop_prod.system.rollback_to_snapshot("mydb.mytest",5440886662709904549)
//      """.stripMargin)

    //查询表 hadoop_prod.mydb.mytest 数据,已经是历史数据
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest
//      """.stripMargin).show(100)

    //10.合并Iceberg表的数据文件
    // 1) 首先向表 mytest 中插入一批数据，将数据写入到表mytest中
//    import spark.implicits._
//    val df: DataFrame = spark.read.textFile("D:\\2018IDEA_space\\Iceberg-Spark-Flink\\SparkIcebergOperate\\data\\nameinfo")
//      .map(line => {
//        val arr: Array[String] = line.split(",")
//        (arr(0).toInt, arr(1), arr(2).toInt)
//      }).toDF("id","name","age")
//    df.writeTo("hadoop_prod.mydb.mytest").append()

    //2) 合并小文件数据,Iceberg合并小文件时并不会删除被合并的文件，Compact是将小文件合并成大文件并创建新的Snapshot。
    // 如果要删除文件需要通过Expire Snapshots来实现,targetSizeInBytes 指定合并后的每个文件大小
//    val conf = new Configuration()
//    val catalog = new HadoopCatalog(conf,"hdfs://mycluster/sparkoperateiceberg")
//    val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
//    Actions.forTable(table).rewriteDataFiles().targetSizeInBytes(1024)//1kb
//      .execute()


    //11.删除历史快照,历史快照是通过ExpireSnapshot来实现的，设置需要删除多久的历史快照
//    val conf = new Configuration()
//    val catalog = new HadoopCatalog(conf,"hdfs://mycluster/sparkoperateiceberg")
//    val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
//    table.expireSnapshots().expireOlderThan(1640066148470L).commit()

    //创建Iceberg 表 mytest2 ,测试SQL 方式保留快照
//    spark.sql(
//      """
//        |create table hadoop_prod.mydb.mytest2 (id int,name string,age int) using iceberg
//      """.stripMargin)
//    //执行以下SQL 语句
//    spark.sql("insert into hadoop_prod.mydb.mytest2 values (1,'zs',18)")
//    spark.sql("insert into hadoop_prod.mydb.mytest2 values (2,'ls',19)")
//    spark.sql("insert into hadoop_prod.mydb.mytest2 values (3,'ww',20)")
//    spark.sql("insert into hadoop_prod.mydb.mytest2 values (4,'ml',21)")
//    spark.sql("insert into hadoop_prod.mydb.mytest2 values (5,'t1',22)")

    //删除早于当前时间戳的快照，但保留最近2个快照
    spark.sql(
      """
        |call hadoop_prod.system.expire_snapshots ('mydb.mytest2')
      """.stripMargin)


    //查看表快照信息
//    spark.sql(
//      """
//        |select * from hadoop_prod.mydb.mytest.snapshots
//      """.stripMargin).show(false)

//    spark.sql("""select * from hadoop_prod.mydb.mytest""").show()
  }

}
